package com.sjztdz.agency_network;

import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import okhttp3.*;

import java.io.*;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.google.common.util.concurrent.*;

public class Main {

    private static Map<String,Object> map = new HashMap<>();
    //每个文件中的行数 uat 200 prd 100000
    private static final long SPLIT_NUM = 300;
    //多线程情况下每秒请求书不要超过的qps uat 5.0 prd 30.0
    private static final RateLimiter GLOBAL_RATE_LIMITER = RateLimiter.create(3.0);

    //20个http请求线程池 超过5分钟不活跃自动销毁 uat 10 prd 20
    private static final OkHttpClient HTTP_CLIENT = new OkHttpClient.Builder()
            .connectTimeout(60, TimeUnit.SECONDS)
            .readTimeout(30, TimeUnit.SECONDS)
            .connectionPool(new ConnectionPool(10, 5, TimeUnit.MINUTES)) // 可选：自定义连接池
            .build();


    public static void main(String[] args) throws InterruptedException {
//        String fuc = "3";
        String fuc = args[0];
        if("1".equals(fuc)){
//            String baseFile = "D:\\youzheng\\代理网点\\2.csv";
                    String baseFile = args[1];
            System.out.println("开始执行某个单独文件，合成后会放在当前目录的merge文件夹里");
            runOneFile(baseFile);
        } else if ("3".equals(fuc)) {
            //合并某个目录下的所有文件
            System.out.println("开始执合并某个目录下的所有文件");
            String mubiao_dir = args[1];
            String mubiao_file = args[2];

//            String mubiao_dir = "D:\\youzheng\\代理网点\\merge";
//            String mubiao_file = "D:\\youzheng\\代理网点\\合并结果.csv";

            try{
                mergeCsvFiles(mubiao_dir,mubiao_file);
            } catch (IOException e) {
                System.out.println("合并文件出现问题,"+e.getMessage());
                throw new RuntimeException(e);
            }
        }else {
            System.out.println("开始拆分文件进行数据获取以及合并");
//            String baseFile = "D:\\youzheng\\代理网点\\2.csv";
            String baseFile = args[1];
            System.out.println(baseFile);
            Path path = Paths.get(baseFile);
            Path parent = path.getParent();
            String temp; //拆分文件的目录
            String temp_merge; //拆分文件的目录
            //合成的目录
            temp_merge = parent.toString() + File.separator + "merge";
            if(Files.exists(Paths.get(temp_merge))){
                System.out.println(temp_merge+"：文件已经存在,尝试创建新合并目录");
                temp_merge = parent.toString() + File.separator + "merge" + System.currentTimeMillis();
                if(Files.exists(Paths.get(temp_merge))){
                    System.out.println("结束！！！创建合并目录失败" + temp_merge);
                    return;
                }
            }
            //拆分的目录
            temp = parent.toString() + File.separator + "split";
            Path path_temp = null;
            path_temp = Paths.get(temp);
            if(Files.exists(path_temp)) {
                System.out.println(temp+"：文件已经存在,尝试创建新分割目录");
                temp = parent.toString() + File.separator + "split" + System.currentTimeMillis();
                path_temp= Paths.get(temp);
                if(Files.exists(path_temp)) {
                    System.out.println("结束！！！创建分割目录失败" + temp);
                    return;
                }
            }
            try {
                Files.createDirectories(path_temp);
                Files.createDirectories(Paths.get(temp_merge));
            } catch (IOException e) {
                System.out.println("创建文件路径失败");
                return;
            }
            //拆分子文件
            String path_temp_str = path_temp.toString();
            splitBigFile(baseFile,path_temp_str);
            //获取子文件列表
            List<Path> paths = listOnlyFiles(path_temp_str);
            //创建线程池
            //计数器 保证线程池完全结束才继续运行主线程的逻辑
            CountDownLatch latch = new CountDownLatch(paths.size());
            ThreadPoolExecutor pool = new ThreadPoolExecutor(
                    2,
                    3,
                    10L,
                    TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(paths.size()-3), //永远不会触发拒绝策略 并且能触发最大线程数
                    new ThreadPoolExecutor.CallerRunsPolicy()
            );
            //文件交给线程池
            for(Path p : paths) {
                try{
                    //匿名函数 隐式集成了Runnable接口
                    String finalTemp_merge = temp_merge;
                    pool.execute(() -> {
                        try{
                            coreTask(p, finalTemp_merge);
                        }finally {
                            latch.countDown();
                        }
                    });
                } catch (Exception e) {
                    System.out.println("当先线程出现异常了，处理的文件是："+p.toString());
                }
            }
            //等待所有线程都运行完
            latch.await();
            //等待所有线程运行完后关闭线程池
            pool.shutdown();
            System.out.println("线程执行完毕——————————————————————————");
            //合并文件
            String merge_file = path.toString().replace(".csv","_merge.csv");
            try{
                mergeCsvFiles(temp_merge,merge_file);
                System.out.println("合并文件完成:"+merge_file);
            }catch (Exception e){
                System.out.println("合并文件失败");
            }
        }
    }

    /**
     * 跑某个指定的文件
     */
    public static void runOneFile(String file_path){
        Path path = Paths.get(file_path);
        Path parent = path.getParent();
        String temp_merge; //拆分文件的目录
        //合成的目录
        temp_merge = parent.toString() + File.separator + "merge";
        if(!Files.exists(Paths.get(temp_merge))){
            try{
                Files.createDirectories(Paths.get(temp_merge));
            } catch (IOException e) {
                System.out.println("创建目录失败,停止运行");
                throw new RuntimeException(e);
            }
        }
        System.out.println("开始转换");
        coreTask(path, temp_merge);
        System.out.println("转换完成，存放目录："+temp_merge);
    }

    /**
     * 获取目标目录的所有文件
     * @param dirPath
     * @return
     * @throws Exception
     */
    public static List<Path> listOnlyFiles(String dirPath) {
        try (Stream<Path> stream = Files.list(Paths.get(dirPath))) { // 只遍历一层
            return stream.filter(Files::isRegularFile)   // 只要普通文件
                    .collect(Collectors.toList());
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }
        return new ArrayList<>();
    }

    /**
     * 把文件先读取到内存然后一次性写入 io次数少 内存要求高
     * 还有一个办法就是一行行的读取后然后写入 io次数多 内存要求低
     * @param file
     * @param output
     * @return
     * @throws FileNotFoundException
     */
    public static boolean splitBigFile(String file, String output) {
        File fileObject = new File(file);
        File outputDir = new File(output);
        if(!fileObject.exists()) return false;
        String fileName = fileObject.getName().replace(".csv", "");
        long length = fileObject.length();
        double mbDecimal = length / 1_000_000.0;
        System.out.println("当前文件大小："+ mbDecimal + "M");
        List<String> list = new LinkedList<>();
        int fileIndex = 0;
        if(fileObject.length()>0){
            boolean firstLine = true;
            String line;
            try(InputStreamReader reader = new InputStreamReader(new FileInputStream(fileObject),StandardCharsets.UTF_8);
                BufferedReader bufferedReader = new BufferedReader(reader)){
                while ((line = bufferedReader.readLine())!=null){
                    if(firstLine){ //跳过第一行 跳过表头
                        firstLine = false;
                        continue;
                    }
                    System.out.println(line);
                    list.add(line);
                    if(list.size() % SPLIT_NUM == 0){
                        //当存到执行行数数据以后开始写成一个文件
                        fileIndex++;
                        writeOneFile(outputDir, fileName, fileIndex, list);
                        //写完以后情况内存数据
                        list.clear();
                    }
                }
                //如果有剩余的文件也要写入
                if (!list.isEmpty()) {
                    fileIndex++;
                    writeOneFile(outputDir, fileName, fileIndex, list);
                }
                System.out.println("文件分割完成！共生成 " + fileIndex + " 个子文件，输出目录：" + outputDir.getAbsolutePath());
                return true;
            } catch (IOException e) {
                return false;
            }
        }
        return false;
    }
    /**
     * 把 list 里的数据一次性写成单个文件
     */
    public static void writeOneFile(File outDir, String baseName, int index, List<String> data) throws IOException {
        File outFile = new File(outDir, baseName + "_" + String.format("%05d", index) + ".csv");
        try (BufferedWriter writer = new BufferedWriter(
                new OutputStreamWriter(new FileOutputStream(outFile), StandardCharsets.UTF_8))) {
            for (String row : data) {
                writer.write(row);
                writer.newLine();
            }
        }
    }


    /**
     * 线程池里提交的核心任务流程
     * @param path
     * @param merge_dir
     */
    public static void coreTask(Path path,String merge_dir){
        String thread_name = Thread.currentThread().getName();
        System.out.println("当前线程:" + thread_name+"  执行文件:"+path.toString());
        String pathList = (String)map.get(thread_name);
        pathList += path.toString() + "\n";
        map.put(thread_name,pathList);
        String name = path.getFileName().toString();
        //用来存储调用完接口后的数据的csv文件路径
        String file_path = merge_dir + File.separator + name;
        File file = new File(file_path);
        String line;
        String new_line;
        //流式读取数据 写入文件 只打开文件一次
        try (BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(file_path), StandardCharsets.UTF_8));
             FileInputStream inputStream = new FileInputStream(String.valueOf(path));
             InputStreamReader reader = new InputStreamReader(inputStream, StandardCharsets.UTF_8);
             BufferedReader br = new BufferedReader(reader)) {
            //当读取到数据时候调用接口
            while ((line = br.readLine()) != null) {
                //阻塞直到拿到许可，防止qps超过30
                GLOBAL_RATE_LIMITER.acquire();
                String[] split = line.split(",");
                String address = split[1];
                //调用外部接口获取数据拼接
                JSONArray postMatchDistance = getPostMatchDistance(address);
                if(postMatchDistance == null){
                    System.out.println("出现问题的文件:"+file_path + "列：" + line);
                    file.delete();//这个文件有出现问题的 重新跑
                    return; //停止执行
                }
                int size = postMatchDistance.size();
                if(size>1){
                    JSONObject data = postMatchDistance.getJSONObject(0);
                    String id = data.getString("yc_inst_code");
                    String post_name = data.getString("yc_inst_name");
                    String distance = data.getString("distance");
                    JSONObject other_data = postMatchDistance.getJSONObject(1);
                    String other_id = other_data.getString("yc_inst_code");
                    String other_post_name = other_data.getString("yc_inst_name");
                    String other_distance = other_data.getString("distance");
                    new_line = line + ","+id + ","+post_name+","+distance+","+other_id+","+other_post_name+","+other_distance;
                } else if (size == 1) {
                    JSONObject data = postMatchDistance.getJSONObject(0);
                    String id = data.getString("yc_inst_code");
                    String post_name = data.getString("yc_inst_name");
                    String distance = data.getString("distance");
                    new_line = line + ","+id + ","+post_name+","+distance;
                }else {
                    new_line = line;
                }
                writer.write(new_line);
                writer.newLine();
            }

        }catch (Exception e) {
            System.out.println(e.getMessage());
        }
    }

    /**
     * 通过具体的地址去匹配当前的地址的经纬度，再调用getPostDistance获取网点的数据
     * @param address
     * @return
     */
    public static JSONArray getPostMatchDistance(String address){
        //根据地址获取经纬度
        StringBuffer sb = new StringBuffer();
//        sb.append("http://10.5.220.158/apps/address/api/match?");
        sb.append("http://localhost:38083/select_trace/test/match?");
        sb.append("app_key").append("=").append("4a871e9f62a74f10b5207a9f22ce11a9").append("&");
        sb.append("addr").append("=").append(address).append("&");
        sb.append("size").append("=").append("1");
        System.out.println(sb.toString());
        for(int i=1;i<4;i++){
            Request request = new Request.Builder()
                    //这里必须手动设置为json内容类型
                    .addHeader("content-type", "application/json")
                    //参数放到链接后面
                    .url(sb.toString())
                    .build();
            try {
                Response res = HTTP_CLIENT.newCall(request).execute();
                String respStr = res.body().string();
                System.out.println(respStr);
                //将响应数据转换json对象
                JSONArray parse = JSONArray.parse(respStr);
                if(parse.isEmpty()){
                    System.out.println("返回结果为空");
                    return null;
                }
                JSONObject jsonObject = parse.getJSONObject(0);
//                JSONObject jsonObject = JSONObject.parseObject(respStr);
//                JSONArray location = jsonObject.getJSONArray("location");
                JSONObject location = jsonObject.getJSONObject("location");
                double lon = location.getDouble("lon");
                double lat = location.getDouble("lat");
                List<Double> loc = new LinkedList<>();
                loc.add(lon);
                loc.add(lat);
                return getPostShop(loc);
            } catch (Exception e) {
                System.out.println("调用当前的地址的经纬度失败:"+address + "失败原因:" + e.getMessage());
                System.out.println("重试，调用");
            }
        }
        System.out.println("三次尝试均没有调用成功");
        return null;
    }

    /**
     * 根据经纬度获取周边网点的数据
     * @param location
     * @return
     */
    public static JSONArray getPostShop(List<Double> location) throws Exception {
        //根据地址获取经纬度
        HashMap<Object, Object> geometry = new HashMap<>();
        System.out.println("地理位置："+location);
        geometry.put("coordinates",location);
        geometry.put("type","Point");
        String jsonString = JSONObject.toJSONString(geometry);
        HttpUrl url = HttpUrl.parse( //http://localhost:38083/select_trace/test/distance  http://10.5.220.158/geodataquery/api/layer-query/buffer-query/distance
                        "http://localhost:38083/select_trace/test/distance")
                .newBuilder()
                .addQueryParameter("app_key", "4a871e9f62a74f10b5207a9f22ce11a9")
                .addQueryParameter("layerId", "8a855d13995bd0260199f0d404a219a9")
                .addQueryParameter("fields", "yc_inst_code,yc_inst_name")
                .addQueryParameter("geometry", jsonString)
                .addQueryParameter("radius", "20000")
                .addQueryParameter("size", "5")
                .build();
        System.out.println(url);
        Request request = new Request.Builder()
                //这里必须手动设置为json内容类型
                .addHeader("Accept", "application/json")
                //参数放到链接后面
                .url(url)
                .build();
        try {
            //获取指定经纬度的周围的网点数据
            Response response = HTTP_CLIENT.newCall(request).execute();
            String respStr = response.body().string();
            System.out.println(respStr);
            JSONObject jsonObject = JSONObject.parseObject(respStr);
            JSONArray data = jsonObject.getJSONArray("data");
            return data;
        } catch (Exception e) {
            System.out.println("调用接口失败："+e.getMessage());
            throw new Exception("调用获取周边数据失败");
        }
    }

    /**
     * 合并一个目录里的所有csv文件
     * @param inputDir 输入目录D:/a/a
     * @param outputFile 输出的文件路径 D:/X.csv
     * @throws IOException
     */
    public static void mergeCsvFiles(String inputDir, String outputFile) throws IOException {
        Path outPath = Paths.get(outputFile);
        Path inDir = Paths.get(inputDir);
        // 确保输出目录存在
        Files.createDirectories(outPath.getParent());
        //只需要打开一次写出的文件避免大量的io

        try (Stream<Path> paths = Files.list(inDir);
             BufferedWriter writer = Files.newBufferedWriter(outPath, StandardCharsets.UTF_8)) {
            //合并写表头
            writer.write("序号,寄件地址,代金网点编号1,代金网点名称1,距离1,代金网点编号2,代金网点名称2,距离2");
            writer.newLine();
            try {
                paths
                        .filter(Files::isRegularFile)
                        .filter(p -> p.getFileName().toString().toLowerCase().endsWith(".csv"))
                        .sorted() // 可选：按文件名排序保证顺序
                        .forEach(path -> {
                            try (BufferedReader reader = Files.newBufferedReader(path, StandardCharsets.UTF_8)) {
                                String line;
                                while ((line = reader.readLine()) != null) {
                                    System.out.println(line);
                                    writer.write(line);
                                    writer.newLine();
                                }
                            } catch (IOException e) {
                                throw new UncheckedIOException(e);
                            }
                        });
            } catch (UncheckedIOException e) {
                throw e.getCause();
            }
        }
    }
}